//
// Ephi - simulation of magnetic fields and particles
// Copyright (C) 2007 Indrek Mandre <indrek(at)mare.ee>
// For more information please see http://www.mare.ee/indrek/
//
// This program is free software; you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation; either version 2 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License along
// with this program; if not, write to the Free Software Foundation, Inc.,
// 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
//

#include "threading.hpp"

Mutex::Mutex ()
{
#ifdef PTHREADS
  pthread_mutexattr_t attr;
  pthread_mutexattr_init (&attr);
  pthread_mutexattr_settype (&attr, PTHREAD_MUTEX_RECURSIVE);
  pthread_mutex_init (&m, &attr);
  pthread_mutexattr_destroy (&attr);
#endif
}

Mutex::~Mutex ()
{
#ifdef PTHREADS
  pthread_mutex_destroy (&m);
#endif
}

void Mutex::lock ()
{
#ifdef PTHREADS
  pthread_mutex_lock (&m);
#endif
}

void Mutex::unlock ()
{
#ifdef PTHREADS
  pthread_mutex_unlock (&m);
#endif
}

bool Mutex::trylock ()
{
#ifdef PTHREADS
  return pthread_mutex_trylock (&m) == 0;
#else
  return true;
#endif
}

Condition::Condition (Mutex& m) : m(m)
{
#ifdef PTHREADS
  pthread_cond_init (&cond, 0);
#endif
}

Condition::~Condition ()
{
#ifdef PTHREADS
  pthread_cond_destroy (&cond);
#endif
}

void Condition::wait ()
{
#ifdef PTHREADS
  pthread_cond_wait (&cond, &m.m);
#endif
}

void Condition::signal ()
{
#ifdef PTHREADS
  pthread_cond_signal (&cond);
#endif
}

void Condition::broadcast ()
{
#ifdef PTHREADS
  pthread_cond_broadcast (&cond);
#endif
}

TaskManager::TaskManager (int threadcount) : work_cond(lock), ready_cond(lock)
{
  queue = complete = 0;
  terminate = false;
  working = 0;
#ifdef PTHREADS
  threads.resize (threadcount);
  for ( size_t i = 0; i < threads.size(); i++ )
      pthread_create (&threads[i], 0, &TaskManager::run_thread, this);
#endif
}

void TaskManager::addTask (Task *task)
{
  AutoMutex m(lock);
  task->next = queue;
  queue = task;
  work_cond.signal ();
}

Task *TaskManager::run()
{
#ifdef PTHREADS
  AutoMutex m(lock);
  while (1)
    {
      if ( complete )
        {
          Task *ret = complete;
          complete = complete->next;
          return ret;
        }

      // all done
      if ( !queue && working == 0 )
          return 0;

      // waiting for something to complete
      ready_cond.wait();
    }
#else
  Task *task = queue;
  if ( task )
    {
      queue = queue->next;
      task->execute();
    }
  return task;
#endif
}

TaskManager::~TaskManager ()
{
  terminate = true;
#ifdef PTHREADS
  {
  AutoMutex m(lock);
  work_cond.broadcast ();
  }
  for ( size_t i = 0; i < threads.size(); i++ )
      pthread_join (threads[i], (void **)0);
#endif
}

void *TaskManager::run_thread (void *ptr)
{
  TaskManager& mgr = *(TaskManager*)ptr;
  Task *task = 0;
  while (true)
    {
      {
      AutoMutex m(mgr.lock);

      if ( mgr.terminate )
          break;

      if ( task )
        {
          task->next = mgr.complete;
          mgr.complete = task;
          mgr.working--;
          mgr.ready_cond.signal ();
          task = 0;
        }

      if ( !mgr.queue )
          mgr.work_cond.wait();

      if ( mgr.queue )
        {
          mgr.working++;
          task = mgr.queue;
          mgr.queue = task->next;
        }
      }

      if ( task )
          task->execute();
    }
  return 0;
}

